dnl m4 stuff to allow the injection of code
include(FileLoaderFunctions.m4)
dnl m4_include(SQLite.m4)

m4_include(FileLoaderDefs.m4)


#include <assert.h>
#include <sys/mman.h>
#include <stdio.h>
#include <stdlib.h>

#include "Errors.h"
#include "DataTypes.h"
#include "FileLoader.h"
#include "PageConversions.h"
#include "DiskArray.h"
#include "MetadataDB.h"
#include <iostream>
#include <vector>
#include <algorithm>
using namespace std;

class objectHolder{
        public:
              //  INT indicies;
             //   FLOAT value;

  IDENTIFIER partkeyData;                                   
  IDENTIFIER supkeyData;                            
  INT linenumberData;                                 
  FLOAT quantityData;                               
  FLOAT extendedpriceData;                                    
  FLOAT discountData;                              
  FLOAT taxData;                              
  CHAR<1> returnflagData;                             
  CHAR<1> linestatusData;                            
  DATE shipdateData;                           
  DATE commitdateData;                               
  DATE receiptdateData;                         
  CHAR<25> shipinstructData;                           
  CHAR<10> shipmodeData;                        
  long int commentData;
  char commentContent;




       friend  bool operator< (const objectHolder& A,const  objectHolder& B);
       friend  bool operator> (const objectHolder& A, const objectHolder& B);
};
bool operator< (const objectHolder& a,const objectHolder& b){
        return (a.value<b.value);
}
bool operator> (const objectHolder& a,const objectHolder& b){
        return (a.value>b.value);
}


FileLoaderImp::FileLoaderImp(const char* _fileName, EventProcessor& _fileWriter, DistributedCounter* _dCounter):
	chunksProduced(0), dCounter(_dCounter), tuplesHandled(0)
#ifdef  DEBUG_EVPROC
	,EventProcessorImp(true, "FileLoader")
#endif
{
	fileName = strdup(_fileName);
	finished=false;

	fileWriter.CopyFrom(_fileWriter);
	DiskArray& diskArray = DiskArray::GetDiskArray();
	pageSize = diskArray.getPageSize();

	// allocate memory for the buffers

	dnl # M4 CODE
	dnl # declaring all the columns
	m4_foreach(</_C_/>, </COLUMN_TYPES/>, </dnl
		M4_ALLOCMEM_COLUMN(_C_)dnl
	/>)
	dnl # END OF M4 CODE

  el_tuples    = (objectHolder*) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*sizeof(objectHolder)),
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_indiciesData = (INT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

 l_sortedkeyData = (IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedpartkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedsupkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedlinenumberData=(INT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedquantityData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedextendedpriceData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sorteddiscountData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedtaxData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedreturnflagData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedlinestatusData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedcommitdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedreceiptdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipinstructData=(CHAR<25>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipmodeData=(CHAR<10>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

                                                              

 l_sortedcommentData=(long int *) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)),
                PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedcommentContent=(char*) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*256),
                       PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);




	// register messages
	RegisterMessageProcessor(ChunkWritten::type, &ChunkWrittenProc, 1);

	// open the file and get ready to read
	stream = fopen(fileName, "r");
	if ( stream==NULL){
		printf( "File %s could not be opened in FileLoader:", fileName);
		perror(NULL);
	}

	// start the whole thing by sending the first chunk
	FillAndSendChunk();
}

FileLoaderImp::~FileLoaderImp(void){
	dnl # M4 CODE
	dnl # declaring all the columns
	m4_foreach(</_C_/>, </COLUMN_TYPES/>, </dnl
		M4_FREEMEM_COLUMN(_C_)dnl
	/>)
	dnl # END OF M4 CODE

  munmap(l_sortedkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedpartkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedsupkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedlinenumberData, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)));

  munmap(l_sortedquantityData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedextendedpriceData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sorteddiscountData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedtaxData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedreturnflagData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));

  munmap(l_sortedlinestatusData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));

  munmap(l_sortedshipdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedcommitdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedreceiptdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedshipinstructData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)));

  munmap(l_sortedshipmodeData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)));
  munmap(l_sortedcommentData, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)));
  munmap(l_sortedcommentContent,  bytesToPageAlligned(pageSize, 2097152*256));


	free(fileName);
	fclose(stream);
}




bool sortByValue(const objectHolder&A, const objectHolder& B){
        return (A.value < B.value);
}

void FileLoaderImp::Done(void){
	/* disclaimer, all of this is in some part pseudocode so ill need help translating it into C
	
	FILE * fileNames;
	fileNames = fopen("fileNames.txt");
	
	vector<FILE *> filePointers = new vector<FILE *>();
	while(!feof(fileNames)) {
		string temp;
		fgets(temp, 20, fileNames);
		
		FILE * pointerr;
		pointerr = fopen(temp);
		
		filePointers.add(pointerr);
	}
	
	int numRuns = filePointers.size();
	int tuplesPerRun = maxTuples/(numRuns+1);
	
	minHeap heap = new MinHeap()
	
	//need some sort of way to keep track of which run each tuple is from
	//initial fill of heap
	
	//make an objectSorter object that has a FILE* and refers to the FILE*'s current objectHolder
	//add the objectSorter to minHeap and heap that, then copy off the top value objectHolder
	//and then increments the FILE* to the next objectHolder
	//do the initial heaping
	for(int i = 0; i < filePointers.size(); i++) {
		objectSorter temp = new objectSorter(filePointers[i]);
		heap.add(temp);
	}
		
	while(filePointers.size() > 0) {
		fout << heap.min();
		if(feof(++(heap.min().File)) { //increment to next objectHolder from last min and checks if its @ eof
			filePointers.remove(heap.min()); //if it is at eof remove pointer from list of pointers and from heap
			heap.remove(heap.min()); 
		}
		reheap();
	}
	*/

	cout <<  "Producer " << fileName << "DONE\n";
	cout << "Have handled :" << tuplesHandled << " Tuples" << endl;
	
	dnl SQL_OPEN_DATABASE( GetMetadataDB() );
	
	// get the relationID
	dnl SQL_STATEMENT_SCALAR(</"SELECT relID FROM Relations WHERE relName='%s';"/>,
	dnl	relID, int, </relName/>);

	// update the freeChunkID of the relation
	dnl SQL_STATEMENTS_NOREZ(</"UPDATE Relations SET freeChunkID=%ld WHERE relName='%s';"/>,
	dnl	</(unsigned long)numChunks, relName/>);
	

	if (dCounter->Decrement(1) == 0){
		// this is the last producer, kill the writer
		printf("I am the last Generator. I'm killing the Writer\n");
		
		// this is the last producer, kill the adder
		DieMessage_Factory(fileWriter);

		sleep(1);
		KillEvProc(fileWriter);
	}
}



//int compare(const void* a, const void *b){ return (*(int*)a-*(int*)b);}

void FileLoaderImp::FillAndSendChunk(void){
	int noTuples=0;
	char* curr;
	char* next;

	size_t lineSize; // the size of the line
	char* buffer = NULL; // will be resized by getline

	dnl # M4 CODE
	dnl # declaring variables for next position for variable types
	m4_foreach(</_C_/>, </COLUMN_TYPES/>, </dnl
		M4_COL_DEFINE_CURR(_C_)dnl
	/>)
	dnl # END OF M4 CODE

	while(true){
		// get next line of text
		// if eof break
		size_t lineRead = getline(&buffer, &lineSize, stream);
		if ( lineRead == -1){
			finished=true; // we read the whole file
			break; // we reached the end of file
		}

		next = buffer;
		
		//cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;

		dnl # M4 CODE
		dnl # declaring all the columns
		m4_foreach(</_C_/>, </COLUMN_TYPES/>, </dnl
			// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != 'SEPARATOR' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			M4_COL_FROM_TOKEN(_C_, noTuples, curr)dnl
		/>)
		dnl # END OF M4 CODE

		noTuples++;
		tuplesHandled++;

		if (noTuples>=TUPLES_PER_CHUNK)
			break;
	}

vector<objectHolder> listOfData;
objectHolder objectIdent;

for(int i = 0; i < noTuples; i++){

        objectIdent.value = *(l_taxData+i);
        objectIdent.indicies = i;
	cout << " taxData" << *(l_taxData+i) << endl;
        listOfData.push_back(objectIdent);
}

        sort(listOfData.begin(), listOfData.end(), sortByValue);

int j = 0;
while(j < noTuples){

        *(l_indiciesData+j) = *(l_linenumberData+listOfData[j].indicies);
        *(l_sortedkeyData+j) = *(l_orderkeyData+listOfData[j].indicies);

        *(l_sortedpartkeyData) = *(l_partkeyData+listOfData[j].indicies); // column l_partkey array                                    
        *(l_sortedsupkeyData) = *(l_supkeyData+listOfData[j].indicies); // column l_supkey array                              
        *(l_sortedlinenumberData) = *(l_linenumberData+listOfData[j].indicies); // column l_linenumber array                                     
        *(l_sortedquantityData) = *(l_quantityData+listOfData[j].indicies); // column l_quantity array                               
        *(l_sortedextendedpriceData) = *(l_extendedpriceData+listOfData[j].indicies); // column l_extendedprice array                                     
        *(l_sorteddiscountData) = *(l_discountData+listOfData[j].indicies); // column l_discount array                               
        *(l_sortedtaxData) = *(l_taxData+listOfData[j].indicies); // column l_tax array                                 
        *(l_sortedreturnflagData) = *(l_returnflagData+listOfData[j].indicies);
        *(l_sortedlinestatusData) = *(l_linestatusData+listOfData[j].indicies);
        *(l_sortedshipdateData) = *(l_shipdateData+listOfData[j].indicies);
        *(l_sortedcommitdateData) = *(l_commitdateData+listOfData[j].indicies);
        *(l_sortedreceiptdateData) = * (l_receiptdateData+listOfData[j].indicies); // column l_receiptdate array                                  
        *(l_sortedshipinstructData) =   *(l_shipinstructData+listOfData[j].indicies);
        *(l_sortedshipmodeData) =  *(l_shipmodeData+listOfData[j].indicies);

   	*(l_sortedcommentData) = *(l_commentData+listOfData[j].indicies);
        *(l_sortedcommentContent) = *(l_commentContent+listOfData[j].indicies);

	cout << "value: " << listOfData[j].value << " index: " << listOfData[j].indicies <<  "sorted " << *(l_sortedkeyData+j) << endl;
j++;
}
	l_linenumberData = l_sortedlinenumberData;
	l_orderkeyData = l_sortedkeyData;
	l_partkeyData = l_sortedpartkeyData;
	l_supkeyData = l_sortedsupkeyData;
	l_extendedpriceData = l_sortedextendedpriceData;
	l_discountData = l_sorteddiscountData;
	l_taxData = l_sortedtaxData;
	l_returnflagData = l_sortedreturnflagData;
	l_linestatusData = l_sortedlinestatusData;
	l_shipdateData = l_sortedshipdateData;
	l_commitdateData = l_sortedcommitdateData;
	l_receiptdateData =l_sortedreceiptdateData;
	l_commitdateData = l_sortedcommitdateData;
	l_receiptdateData = l_sortedreceiptdateData;
	l_shipinstructData = l_sortedshipinstructData;
	l_shipmodeData = l_sortedshipmodeData;	
	l_commentData = l_sortedcommentData;
	l_commentContent = l_sortedcommentContent;
	// deallocate the buffer (not needed anymore)
	free(buffer);

	// we might be finished and not produce any tuples
	// if so we do not want to sent a request to write an empty chunk to the disk
	if (finished && noTuples == 0){
		Done();
		return;
	}

//temp disk



//merge


//send


	// send the request
	ColWriteDescriptionContainer cols;

	dnl # M4 CODE
	dnl # forming columns information
	m4_foreach(</_C_/>, </COLUMN_TYPES/>, </dnl
		M4_COL_DESCRIPTION(_C_, cols, noTuples)dnl
	/>)
	dnl # END OF M4 CODE

	// if we do not clone the interface, the method swaps the content out
	EventProcessor myInterfClone;
	myInterfClone.CopyFrom(myInterface);
	ChunkWrite_Factory( fileWriter, chunksProduced, noTuples, myInterfClone, cols);

	cout << "Loader of  " << fileName <<" : chunk " << chunksProduced << " out" << endl;

	chunksProduced++;
}

MESSAGE_HANDLER_DEFINITION_BEGIN(FileLoaderImp, ChunkWrittenProc, ChunkWritten){

	cout << "Loader of  " << evProc.fileName << " : chunk " << evProc.chunksProduced << " BACK " << endl;

	if (!evProc.finished){
		evProc.FillAndSendChunk();
	} else {
		evProc.Done();
	}
}MESSAGE_HANDLER_DEFINITION_END
